package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN071;
import com.surfilter.massdata.spark.bean.SAN081;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

@SuppressWarnings("serial")
public class NetizenAnalyzCityOperTask_New implements ISparkTask {
	private String city;
	private String oper;
	private String iptype;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,
			Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		try{
			// 读取配置数据
			DataFrame dns = dataFrames.get("dws_hour_san081");
			SQLContext sqlContext = dns.sqlContext();
			String dayStr = commandMap.get("-d");
			Date date = DateUtil.getExecDate(new Date(), dayStr);			
			// 网民分布统计-地市网民访问量
			//JavaRDD<Row> rdd=dns.toJavaRDD();
			DataFrame city = getCityDF(date,dns);
			list.add(new OutQueueEntity(this.city, city));

			// 网民分布统计-运营商网民访问量
			DataFrame oper = getOperatorDF(date,dns);
			list.add(new OutQueueEntity(this.oper, oper));

			// 网民分布统计-网民访问量分类统计
			/*DataFrame iptype = getIpTypeDF(sqlContext, date,dns.toJavaRDD());
			list.add(new OutQueueEntity(this.iptype, iptype));*/
			CommonUtils.deleteTaskTableData("SAN081", date, "COUNT_AREA_SUM", 0, "day");
			CommonUtils.deleteTaskTableData("SAN081", date, "COUNT_OPERATORS_SUM", 0, "day");
			//CommonUtils.deleteTaskTableData("SAN071", date, "COUNT_IP_TYPE", 0, "day");
		}catch(Exception e){
			SysLog.error(e.getMessage());
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("NetizenAnalyzCityOperTask_New:exectime: "+min+" min............");
		
		return list;
	}


	private DataFrame getCityDF(final Date date,DataFrame dataSource) {
		
		String sql = "select sum(count_value)as count,BUSS_VALUE  from dws_hour_san081_temp where BUSS_TYPE='COUNT_AREA_SUM' and sta_range=0 GROUP BY BUSS_VALUE";
		DataFrame df = dataSource.sqlContext().sql(sql);
		
		JavaRDD<SAN081> cityRDD = df.toJavaRDD().map(new Function<Row, Map<String,String>>() {

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("count").toString());
				map.put("city", row.getAs("BUSS_VALUE").toString());
				return map;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN081>() {

			@Override
			public Iterable<SAN081> call(Iterator<Map<String, String>> it) throws Exception {

				List<SAN081> list = new ArrayList<SAN081>();
				try{
					while(it.hasNext()){
						Map<String, String> map = it.next();
						String city = String.valueOf(map.get("city"));
						long count = Long.parseLong(String.valueOf(map.get("count")));
						if(StringUtils.isNotBlank(city)){
							SAN081 sa = new SAN081();
							
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_AREA_SUM");
							sa.setBuss_value(city);
							sa.setCount_value(count);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		
		
		DataFrame resultDf = dataSource.sqlContext().createDataFrame(cityRDD, SAN081.class);
		return resultDf;
	}

	
	
	private DataFrame getOperatorDF(final Date date,DataFrame dataSource) {
		
		String sql = "select sum(count_value)as count,BUSS_VALUE  from dws_hour_san081_temp where BUSS_TYPE='COUNT_OPERATORS_SUM' and sta_range=0 GROUP BY BUSS_VALUE";
		DataFrame df = dataSource.sqlContext().sql(sql);
		
		JavaRDD<SAN081> operatorRDD = df.toJavaRDD().map(new Function<Row, Map<String,String>>() {

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("count").toString());
				map.put("operators_id", row.getAs("BUSS_VALUE").toString());
				return map;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN081>() {

			@Override
			public Iterable<SAN081> call(Iterator<Map<String, String>> it) throws Exception {

				List<SAN081> list = new ArrayList<SAN081>();
				try{
					while(it.hasNext()){
						Map<String, String> map = it.next();
						String operators_id = String.valueOf(map.get("operators_id"));
						long count = Long.parseLong(String.valueOf(map.get("count")));
						if(StringUtils.isNotBlank(city)){
							SAN081 sa = new SAN081();
							
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_OPERATORS_SUM");
							sa.setBuss_value(operators_id);
							sa.setCount_value(count);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		
		
		DataFrame resultDf = dataSource.sqlContext().createDataFrame(operatorRDD, SAN081.class);
		return resultDf;
	}

	private DataFrame getIpTypeDF(SQLContext sqlContext,final Date date,JavaRDD<Row> rdd) {
		JavaRDD<SAN071> ipTypeRDD = rdd.mapToPair(new PairFunction<Row,String,Long>() {

			@Override
			public Tuple2<String,Long> call(Row map){
				try{
					String ip_type = CommonUtils.valueOf(map.getAs("buss_value"));
					if(StringUtils.isNotBlank(ip_type)){
						long dns_visit_count = Long.parseLong(CommonUtils.valueOf(map.getAs("count_value")));
						return new Tuple2<String,Long>(ip_type,dns_visit_count);
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return new Tuple2<String,Long>("",0L);
			}
		}).reduceByKey(new Function2<Long, Long, Long>() {
			
			@Override
			public Long call(Long v1, Long v2){
				return v1 + v2;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Tuple2<String,Long>>,SAN071>() {

			@Override
			public Iterable<SAN071> call(Iterator<Tuple2<String, Long>> it){
				List<SAN071> list = new ArrayList<SAN071>();
				try{
					while(it.hasNext()){
						Tuple2<String,Long> tuple = it.next();
						String ip_type = tuple._1;
						if(StringUtils.isNotBlank(ip_type)){
							SAN071 sa = new SAN071();
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_IP_TYPE");
							sa.setBuss_value(ip_type);
							sa.setCount_value(tuple._2);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		DataFrame df = sqlContext.createDataFrame(ipTypeRDD, SAN071.class);
		return df;
	}
}